#include "EpollProactor.h"
#include "EpollAsynchIO.h"
#include "Debug.h"
#include "Trace.h"

#include <errno.h>
#include <string.h>
#include <unistd.h>
#include <sys/epoll.h>

int EpollProactorNotify::open (void)
{
    TRACE ("");
    if (this->notification_pipe_.open () == -1)
        return -1;
    // close-on-exec
    ::fcntl (this->notification_pipe_.read_handle (), F_SETFD, FD_CLOEXEC);
    ::fcntl (this->notification_pipe_.write_handle (), F_SETFD, FD_CLOEXEC);

    // nonblock mode
    NDK::set_non_block_mode (this->notification_pipe_.write_handle ());
    NDK::set_non_block_mode (this->notification_pipe_.read_handle ());

    return 0;
}
int EpollProactorNotify::close (void)
{
    return this->notification_pipe_.close ();
}
int EpollProactorNotify::notify (void)
{
    TRACE ("");
    return NDK::write (this->notification_pipe_.write_handle (),
            "1234",
            4) > 0 ? 0 : -1; 
}
void EpollProactorNotify::handle_read_stream (void)
{
    TRACE ("");
    int result = 0;
    static char buff[8] = {0};
    while (1)
    {
        result = NDK::read (this->notification_pipe_.read_handle (),
                buff,
                sizeof (buff));
        if (result == -1 && errno == EAGAIN)
            break;
        else if (result > 0)
        {
            if (static_cast<size_t>(result) < sizeof (buff))
                break;
        }
    }
}
NDK_HANDLE EpollProactorNotify::notify_handle (void) const
{
    return this->notification_pipe_.read_handle ();
}
// -------------------------------------------------------------------------
EpollProactorWriteTask::EpollProactorWriteTask ()
: free_list_head_ (0)
, free_list_tail_ (0)
, job_list_head_ (0)
, job_list_tail_ (0)
, free_count_ (0)
, size_ (0)
, task_free_ (1)
{
    TRACE ("");
}
EpollProactorWriteTask::~EpollProactorWriteTask ()
{
    TRACE ("");
    Guard_T<ThreadMutex> g (this->free_list_mutex_);
    if (this->free_list_head_)
    {
        MessageBlock *mb = 0;
        for (this->free_list_tail_ = 0; this->free_list_head_ != 0; )
        {
            mb = this->free_list_head_;
            this->free_list_head_ = this->free_list_head_->next ();
            mb->clean ();  //
            delete mb;
        }
    }
    this->free_list_head_ = 0;
    this->free_count_ = 0;
    this->size_ = 0;
    this->free_list_tail_ = 0;
}
MessageBlock *EpollProactorWriteTask::get_message_block (void)
{
    Guard_T<ThreadMutex> g (this->free_list_mutex_);
    return this->get_message_block_i ();
}
MessageBlock *EpollProactorWriteTask::get_message_block_i (void)
{
    if (this->free_count_ == 0)
    {
        // prealloc members
        this->free_list_head_ = new MessageBlock (sizeof (JobInfo));
        if (this->free_list_head_ == 0) return 0;
        MessageBlock *itor = this->free_list_head_;
        for (int i = 0; i < 512; ++i)
        {
            MessageBlock *mb = new MessageBlock (sizeof (JobInfo));
            if (mb)
            {
                itor->next (mb);
                itor = itor->next ();
                ++this->free_count_;
            }
        }
        this->free_list_tail_ = itor;
        this->size_ += this->free_count_;
    }
    MessageBlock *mb = this->free_list_head_;
    this->free_list_head_ = this->free_list_head_->next ();
    if (this->free_list_head_ == 0) 
        this->free_list_tail_ = 0;
    --this->free_count_;
    mb->next (0);
    return mb;
}
void EpollProactorWriteTask::unget_message_block (MessageBlock *mb)
{
    Guard_T<ThreadMutex> g (this->free_list_mutex_);
    this->unget_message_block_i (mb);
}
void EpollProactorWriteTask::unget_message_block_i (MessageBlock *mb)
{
    if (mb == 0) return ;
    mb->next (0);
    memset (mb->base (), 0, sizeof (JobInfo));
    ++this->free_count_;
    if (this->free_list_tail_ == 0) 
    {
        this->free_list_tail_ = mb;
        this->free_list_head_ = mb;
    }else
    {
        this->free_list_tail_->next (mb);
        this->free_list_tail_ = this->free_list_tail_->next ();
    }
}
int EpollProactorWriteTask::svc (void)
{
    const TimeValue *timeout = 0;
    while (1)
    {
        MessageBlock *mb = 0;
        if (this->task_free_) 
            timeout = 0;
        else
            timeout = &TimeValue::zero;
        // 1. 
        if (this->getq_n (mb, 1024, timeout) == -1 
                || mb == 0)
            continue;
        // 2. 
        this->check_message_block (mb);

        // 3.
        this->execute_write_operation ();
    }
    return 0;
}
void EpollProactorWriteTask::check_message_block (MessageBlock *mb_list)
{
    MessageBlock *mb = mb_list;
    while (mb)
    {
        if (mb->data_type () == AIO_DEL_WRITE_JOB)
        {
            JobInfo *djob = reinterpret_cast<JobInfo *>(mb->base ());
            MessageBlock *itor = this->job_list_head_;
            MessageBlock *prev = 0;
            while (itor)
            {
                JobInfo *job = reinterpret_cast<JobInfo *>(itor->base ());
                // 1. Match the handle.
                if (job->operation->handle () 
                        == djob->handle)
                {
                    MessageBlock *tmp = itor;
                    // Delete this job.
                    if (prev == 0) // The head node.
                    {
                        this->job_list_head_ = itor->next ();
                        itor = this->job_list_head_;
                    }else
                    {
                        prev->next (itor->next ());
                        itor->next (0);
                        itor = prev->next ();
                    }
                    this->unget_message_block (tmp);
                    continue;
                    //break;   !!!!
                }
                // 2 Advance the iterator. !!!
                prev = itor;
                itor = itor->next ();
            }
            MessageBlock *node = mb;
            mb = mb->next ();
            this->unget_message_block (node);
        }else if (mb->data_type () == AIO_ADD_WRITE_JOB)
        {
            if (this->job_list_tail_ == 0 
                    || this->job_list_head_ == 0)
            {
                this->job_list_head_ = mb;
            }else
            {
                this->job_list_tail_->next (mb);
            }
            this->job_list_tail_ = mb;
            mb = mb->next ();
        }
    }
}
int EpollProactorWriteTask::execute_write_operation (void)
{
    //
    MessageBlock *itor = this->job_list_head_;
    MessageBlock *prev = 0;
    while (itor)
    {
        JobInfo *job = reinterpret_cast<JobInfo *>(itor->base ());
        if (job && job->operation && job->operation->execute (job->result) 
                == 1)
            // Write finished.
        {
            if (job->result) 
            {
                if (job->result->complete (0, 0, 0, 0) != 0)
                {
                    job->result->asynch_handler_->handle_close (
                            AsynchHandler::WRITE_MASK);
                }
                delete job->result; // !!!
                job->result = 0;
            }

            // Delete this job.
            if (prev == 0) // The head node.
            {
                this->job_list_head_ = itor->next ();
                this->unget_message_block (itor);
                itor = this->job_list_head_;
            }else
            {
                prev->next (itor->next ());
                itor->next (0);
                this->unget_message_block (itor);
                itor = prev->next ();
            }
            continue;
        }
        prev = itor; 
        itor = itor->next ();
    }

    if (this->job_list_head_ == 0)
    {
        this->task_free_ = 1;
        this->job_list_tail_ = 0;
    }else
        this->task_free_ = 0;
    return 0;
}
// -------------------------------------------------------------------------
EpollProactor::EpollAioFdInfo::EpollAioFdInfo ()
: handle (NDK_INVALID_HANDLE)
, event_mask (0)
, active_mask (0)
, status (FD_CLOSED)
, operation (0)
, prev (0)
, next (0)
{
}
void EpollProactor::EpollAioFdInfo::reset (void)
{
    TRACE ("");
    this->handle = NDK_INVALID_HANDLE;
    this->event_mask	= 0;
    this->active_mask	= 0;
    this->status	= FD_CLOSED;
    this->operation     = 0;
    this->prev		= 0;
    this->next		= 0;
}
// -------------------------------------------------------------------------
EpollProactor::EpollProactor ()
: initialized_ (false)
, epoll_fd_ (NDK_INVALID_HANDLE)
, size_ (0)
, deactivated_ (0)
, events_ (0)
, start_pevents_ (0)
, end_pevents_ (0)
, notify_handler_ (0)
, aio_fd_info_ (0)
, not_finished_list_head_ (0)
, not_finished_list_tail_ (0)
, owner_ (NULL_thread)
, token_ (this)
{	
    TRACE ("");
}
EpollProactor::~EpollProactor ()
{
    TRACE ("");
    this->close ();
}
int EpollProactor::open (int num_of_threads/* = 2*/, int max_handles/* = 0*/)
{
    TRACE ("");
    if (this->initialized_)
        return -1;

    Guard_T<EpollProactorToken> g (this->token_);
    // 1. Set capbility
    this->size_        = max_handles;
    if (this->size_ == 0)
        this->size_ = NDK::max_handles ();
    else
    {
        NDK::set_handle_limit (this->size_);
    }
    if (this->size_ <= 0)
        return -1;

    int result = 0;
    // 2. Allocate event pool.
    this->events_ = new epoll_event[this->size_];
    if (this->events_ == 0)
        result = -1;

    // 3. Allocate result pool.
    if (result != -1)
    {
        this->aio_fd_info_ = new EpollProactor::EpollAioFdInfo[this->size_];
        if (this->aio_fd_info_ == 0)
            result = -1;
    }

    // 4. Open epoll driver.
    if (result != -1)
    {
        this->epoll_fd_ = ::epoll_create (this->size_);
        if (this->epoll_fd_ == NDK_INVALID_HANDLE)
        {
            result = -1;
        }else
            ::fcntl (this->epoll_fd_, F_SETFD, FD_CLOEXEC);
    }

    // 5. Create notifier.
    if (result != -1)
    {
        this->notify_handler_ = new EpollProactorNotify();
        if (this->notify_handler_ == 0
                ||
                this->notify_handler_->open () == -1
                || 
                this->register_handle_i (this->notify_handler_->notify_handle (),
                    0,
                    AsynchHandler::READ_MASK) == -1)
            result = -1;
    }

    // 6. Activate write task.
    if (num_of_threads < 2) result = -1;
    if (result != -1)
    {
        this->write_task_ = new EpollProactorWriteTask();
        if (this->write_task_ == 0 
                || this->write_task_->activate (Thread::THR_JOIN,
                    num_of_threads - 1) == -1)
            result = -1; 
    }
    if (result != -1)
        this->initialized_ = true;
    else
        this->close ();

    return result;
}
int EpollProactor::close (void)
{
    this->initialized_ = false;
    if (this->events_)
    {
        delete [] this->events_;
        this->events_	= 0;
    }
    if (this->aio_fd_info_)
    {
        delete [] this->aio_fd_info_;
        this->aio_fd_info_  = 0;
    }
    if (this->notify_handler_)
    {
        delete this->notify_handler_;
        this->notify_handler_	= 0;
    }
    NDK_OS::close (this->epoll_fd_);
    this->epoll_fd_  = NDK_INVALID_HANDLE;
    return 0;
}
int EpollProactor::notify (void)
{
    TRACE ("");
    return this->notify_handler_->notify ();
}
int EpollProactor::run_event_loop (void)
{
    TRACE ("");
    this->owner (Thread::self ());
    if (this->event_loop_done ())
        return 0;

    int result  = 0;
    while (1)
    {
        result = this->handle_events ();
        if (result == -1 && this->event_loop_done ())
            return 0;
        else if (result == -1)
            return -1;
    }
    return 0;
}
void EpollProactor::end_event_loop (void)
{
    TRACE ("");
    this->deactivated_ = 1;
}
int EpollProactor::event_loop_done (void)
{
    TRACE ("");
    return this->deactivated_;
}
static void polite_sleep_hook (void *) { }
int EpollProactor::handle_events (const TimeValue *timeout/* = 0*/) 
{
    TRACE ("");
    if (Thread::thr_equal (Thread::self (), this->owner_) == 0
            || this->deactivated_)
        return -1;
    int result = this->token_.acquire_read (&polite_sleep_hook, 
            0, 
            timeout);
    if (result != 0)
    {
        this->token_.release ();
        return result;
    }

    result = this->handle_event_i (timeout);
    this->token_.release ();
    return result;
}
int EpollProactor::register_handle (NDK_HANDLE handle, 
        const void *completion_key,
        int mask) 
{
    TRACE ("");
    if (mask == 0) return -1;

    Guard_T<EpollProactorToken> g (this->token_);

    return this->register_handle_i (handle, completion_key, mask);
}
int EpollProactor::register_handle_i (NDK_HANDLE handle, 
        const void * /*completion_key*/, 
        int mask)
{
    TRACE ("");
    struct epoll_event ep;
    ::memset (&ep, 0, sizeof (ep));
    if (NDK_BIT_ENABLED (mask, AsynchHandler::READ_MASK))
    {
        NDK_SET_BITS (ep.events, EPOLLIN);
    }
#if 0
    if (NDK_BIT_ENABLED (mask, AsynchHandler::WRITE_MASK))
    {
        NDK_SET_BITS (ep.events, EPOLLOUT);
    }
#endif

    NDK_SET_BITS (ep.events, (EPOLLHUP | EPOLLERR));
    //NDK_SET_BITS (ep.events, EPOLLET);

    NDK::set_non_block_mode (handle);

    ep.data.fd = handle;
    int result = ::epoll_ctl (this->epoll_fd_,
            EPOLL_CTL_ADD,
            handle,
            &ep);
    if (result == -1 && errno != EEXIST)
    {
        return -1;
    }
    // Change fd status.
    Guard_T<AIO_FD_INFO_MUTEX_T> g(this->aio_fd_info_mutex_);
    EpollAioFdInfo &info = this->aio_fd_info_[handle];
    info.handle = handle;
    info.event_mask = AsynchHandler::READ_MASK;
    info.status = EpollAioFdInfo::FD_REGISTERED;
    return result;
}
int EpollProactor::handle_event_i (const TimeValue *timeout) 
{
    TRACE ("");
    // 1. Proccess the fds that not be finished I/O.
    if (this->not_finished_list_head_ != 0)
    {
        return this->process_notfinished_fds ();
    }

    // 2. Wait epoll fd event
    if (this->wait_for_ready_fd (timeout) > 0)
        // 3. Proccess new ready fds.
    {
        return this->process_ready_fds ();
    }
    return 0;
}
int EpollProactor::process_notfinished_fds (void)
{
    TRACE ("");
    if (this->not_finished_list_head_ == 0)
        return 0;

    Guard_T<AIO_FD_INFO_MUTEX_T> g(this->aio_fd_info_mutex_);
    int result = 0;
    EpollAioFdInfo *itor = this->not_finished_list_head_;
    while (itor)
    {
        EpollAioFdInfo *node = itor;
        //if (node->status == EpollAioFdInfo::FD_NOT_FINISHED)
        if (node->operation && node->status != EpollAioFdInfo::FD_CLOSED)
        {
            EpollAsynchResult *opt_result = 0;
            if (node->operation->execute (opt_result) == 1)
                // Operation finished, delete this node
            {
                node->status = EpollAioFdInfo::FD_FINISHED;
                if (node->prev == 0) // Head node
                {
                    if (node->next == 0) // Only one
                    {
                        this->not_finished_list_head_ = 0;
                        this->not_finished_list_tail_ = 0;
                    }else
                    {
                        this->not_finished_list_head_->next->prev = 0;
                        this->not_finished_list_head_ = node->next;
                    }
                }else if (node->next == 0)  // Tail node.
                {
                    if (node->prev == 0) // Only one
                    {
                        this->not_finished_list_head_ = 0;
                        this->not_finished_list_tail_ = 0;
                    }else
                    {
                        node->prev->next = 0;
                        this->not_finished_list_tail_ = node->prev;
                    }
                }else  // Middle node
                {
                    node->next->prev = node->prev;
                    node->prev->next = node->next;
                }
            }
            node->operation = 0;  // !!! Rebind at next time.
            ++result;
            // AIO complete. 
            if (opt_result)
            {
                if (opt_result->complete (0, 0, 0, 0) == 0) // !!
                {
                    // Advance iterator !!!
                    itor = itor->next;
                    continue;  // 
                }else
                {
                    opt_result->asynch_handler_->handle_close (
                            AsynchHandler::READ_MASK);
                    // goto END
                }
                // goto END
            }
            // goto END
        } // end off if (node->operation) 

        // END
        node->reset ();
        this->remove_handle_i (node->handle,
                0, 
                AsynchHandler::READ_MASK);
        // Advance iterator !!!
        itor = itor->next;
    }
    return result; 
}
int EpollProactor::process_ready_fds (void)
{
    TRACE ("");
    Guard_T<AIO_FD_INFO_MUTEX_T> g(this->aio_fd_info_mutex_);

    struct epoll_event *& pfd = this->start_pevents_;
    int result = 0;
    //NDK_INF ("proccess ready fds\n");
    while (pfd < this->end_pevents_)
    {
        if (pfd->events == 0)
        {
            ++pfd;  // advance iterator !!!
            continue;
        }
        // Do not report notification pipe
        if (pfd->data.fd == this->notify_handler_->notify_handle ())
        {
            this->notify_handler_->handle_read_stream ();
            ++pfd;  // advance iterator !!!
            continue;
        }

        // = Process active events.
        EpollAioFdInfo &info = this->aio_fd_info_[pfd->data.fd];
        if (NDK_BIT_ENABLED (pfd->events, (EPOLLIN | EPOLLHUP | EPOLLERR)))
        {
            if (info.operation && info.status != EpollAioFdInfo::FD_CLOSED)
            {
                //NDK_INF ("proccess readable fd\n");
                NDK_SET_BITS (info.active_mask, AsynchHandler::READ_MASK);
                EpollAsynchResult *opt_result = 0;
                if (info.operation->execute (opt_result) != 1) 
                    // Operation not finished.
                {
                    info.status = EpollAioFdInfo::FD_NOT_FINISHED;
                    // Push this fd-info to the head of not-finished list.
                    info.next = 0;
                    info.prev = 0;
                    if (this->not_finished_list_tail_ == 0) // The first one.
                    {
                        this->not_finished_list_head_ = &info;
                        this->not_finished_list_tail_ = &info;
                    }else 
                    {
                        this->not_finished_list_tail_->next = &info;
                        info.prev = this->not_finished_list_tail_;
                        this->not_finished_list_tail_ = &info;
                    }
                }
                // AIO complete. 
                info.operation = 0;
                ++result;
                if (opt_result)
                {
                    if (opt_result->complete (0, 0, 0, 0) == 0)
                    {
                        ++pfd;  // advance iterator !!!
                        continue;
                    }else
                    {
                        opt_result->asynch_handler_->handle_close (
                                AsynchHandler::READ_MASK);
                        // goto End.
                    }
                    // goto End.
                }
                // goto End.
            }
            // End.
            info.reset ();
            this->remove_handle_i (pfd->data.fd,
                    0,
                    AsynchHandler::READ_MASK);
        }
        ++pfd;  // advance iterator !!!
    }
    return result;
}
int EpollProactor::wait_for_ready_fd (const TimeValue *timeout)
{
    TRACE ("");
    if (this->start_pevents_ != this->end_pevents_)
        return 1;
    int wait_time = -1L;
    if (timeout != 0)
    {
        wait_time = static_cast<int>(timeout->msec ());
    }
    int nfds = 0;
    do
    {
        nfds = ::epoll_wait (this->epoll_fd_,
                this->events_,
                this->size_,
                wait_time);
    }while (nfds == -1 && errno == EINTR);
    if (nfds > 0)
    {
        this->start_pevents_ = this->events_;
        this->end_pevents_   = this->start_pevents_ + nfds;
    }else
        nfds = 0;
    return nfds;
}
int EpollProactor::start_read (EpollAsynchOpt *opt)
{
    TRACE ("");
    if (opt == 0 || opt->result_ == 0) return -1;

    // Store the result in exact slot.
    NDK_HANDLE handle = opt->handle ();
    if (handle == NDK_INVALID_HANDLE)
        return -1;
    if (handle >= this->size_) return -1; // avoid memory flow over.

    // Must lock here, avoid other threads call read/write operation.
    Guard_T<AIO_FD_INFO_MUTEX_T> g(this->aio_fd_info_mutex_);
    EpollAioFdInfo &info = this->aio_fd_info_[handle];

    if (info.status == EpollAioFdInfo::FD_CLOSED)
    {
        return -1;
    }
    info.operation = opt;
    return 0;
}
int EpollProactor::start_write (EpollAsynchOpt *opt, 
        EpollAsynchResult *result)
{
    MessageBlock *mb = this->write_task_->get_message_block ();
    EpollProactorWriteTask::JobInfo *job = 
        reinterpret_cast<EpollProactorWriteTask::JobInfo *>(mb->base ());
    job->operation = opt;
    job->result = result;
    job->handle = opt->handle ();
    mb->data_type (EpollProactorWriteTask::AIO_ADD_WRITE_JOB);
    this->write_task_->putq (mb);
    return 0;
}
int EpollProactor::cancel_aio (NDK_HANDLE handle,
        int event_mask)
{
    TRACE ("");
    if (handle >= this->size_) return -1; // avoid memory flow over.
    if (handle == NDK_INVALID_HANDLE) return -1;

    if (NDK_BIT_ENABLED (event_mask, AsynchHandler::READ_MASK))
    {
        Guard_T<AIO_FD_INFO_MUTEX_T> g(this->aio_fd_info_mutex_);

        this->aio_fd_info_[handle].event_mask = 0;
        this->aio_fd_info_[handle].operation  = 0;
        this->aio_fd_info_[handle].status     = EpollAioFdInfo::FD_CLOSED;
    }else if (NDK_BIT_ENABLED (event_mask, AsynchHandler::WRITE_MASK))
    {
        MessageBlock *mb = this->write_task_->get_message_block ();
        EpollProactorWriteTask::JobInfo *job =
            reinterpret_cast<EpollProactorWriteTask::JobInfo *>(mb->base ());
        job->handle = handle;
        mb->data_type (EpollProactorWriteTask::AIO_DEL_WRITE_JOB);
        this->write_task_->putq (mb);
        return 0;
    }
    return this->remove_handle (handle, 0, event_mask);
}
int EpollProactor::remove_handle (NDK_HANDLE handle, 
        const void *completion_key,
        int mask) 
{
    TRACE ("");
    Guard_T<EpollProactorToken> g (this->token_);

    return this->remove_handle_i (handle, completion_key, mask);
}
int EpollProactor::remove_handle_i (NDK_HANDLE handle, 
        const void * /*completion_key*/,
        int mask) 
{
    TRACE ("");
    struct epoll_event ep;
    ::memset (&ep, 0, sizeof (ep));

    if (NDK_BIT_ENABLED (mask, AsynchHandler::READ_MASK))
        NDK_SET_BITS (ep.events, EPOLLIN);

    if (NDK_BIT_ENABLED (mask, AsynchHandler::WRITE_MASK))
        NDK_SET_BITS (ep.events, EPOLLOUT);

    if (mask == 0)
        NDK_SET_BITS (ep.events, (EPOLLIN | EPOLLOUT));
    NDK_SET_BITS (ep.events, (EPOLLHUP | EPOLLERR));
    //NDK_SET_BITS (ep.events, EPOLLET);
    //ep.events = (EPOLLIN | EPOLLOUT | EPOLLET);
    ep.data.fd = handle;
    int result = ::epoll_ctl (this->epoll_fd_,
            EPOLL_CTL_DEL,
            handle,
            &ep);
    if (result == -1)
    {
        if (errno == ENOENT)  // already removed, it is OK
            return 0;
    }
    return result;
}
AsynchReadStreamImpl *EpollProactor::create_asynch_read_stream (void) 
{
    TRACE ("");
    AsynchReadStreamImpl *impl= 0;
    impl = new EpollAsynchReadStream (this);
    return impl;
}
AsynchReadStreamResultImpl *EpollProactor::create_asynch_read_stream_result (
        const AsynchHandler *handler,
        NDK_HANDLE handle, 
        MessageBlock *mb, 
        size_t bytes_to_read) 
{
    TRACE ("");
    AsynchReadStreamResultImpl *impl = 0;
    impl = new EpollAsynchReadStreamResult(handler,
            handle,
            mb,
            bytes_to_read);
    return impl;
}
AsynchWriteStreamImpl *EpollProactor::create_asynch_write_stream (void) 
{
    TRACE ("");
    AsynchWriteStreamImpl *impl = 0;
    impl = new EpollAsynchWriteStream (this);
    return impl;
}
AsynchWriteStreamResultImpl *EpollProactor::create_asynch_write_stream_result (
        const AsynchHandler *handler,
        NDK_HANDLE handle, 
        MessageBlock *mb, 
        size_t bytes_to_write) 
{
    TRACE ("");
    AsynchWriteStreamResultImpl *impl = 0;
    impl = new EpollAsynchWriteStreamResult(handler,
            handle,
            mb,
            bytes_to_write);
    return impl;
}
AsynchAcceptImpl *EpollProactor::create_asynch_accept (void) 
{
    TRACE ("");
    AsynchAcceptImpl *impl = 0;
    impl = new EpollAsynchAccept (this);
    return impl;
}
AsynchAcceptResultImpl *EpollProactor::create_asynch_accept_result (
        const AsynchHandler *handler,
        NDK_HANDLE listen_handle, 
        MessageBlock *mb, 
        size_t num_to_accept) 
{
    TRACE ("");
    AsynchAcceptResultImpl *impl = 0;
    impl = new EpollAsynchAcceptResult(handler,
            listen_handle,
            mb,
            num_to_accept);
    return impl;
}
